<?php

namespace WebmanGrpc\Stream;

use Google\Protobuf\Internal\Message;
use Generator;

/**
 * 流式处理器
 * 
 * 提供流式处理的内存控制功能
 */
class StreamProcessor
{
    /**
     * 最大内存使用量（字节）
     * @var int
     */
    protected int $maxMemoryUsage;
    
    /**
     * 当前内存使用量（字节）
     * @var int
     */
    protected int $currentMemoryUsage = 0;
    
    /**
     * 批处理大小
     * @var int
     */
    protected int $batchSize;
    
    /**
     * 是否启用内存监控
     * @var bool
     */
    protected bool $memoryMonitoringEnabled = true;
    
    /**
     * 构造函数
     * 
     * @param array $options 配置选项
     */
    public function __construct(array $options = [])
    {
        $defaultOptions = [
            'max_memory_usage' => 50 * 1024 * 1024,  // 50MB
            'batch_size' => 100,
            'memory_monitoring_enabled' => true,
        ];
        
        $options = array_merge($defaultOptions, $options);
        
        $this->maxMemoryUsage = $options['max_memory_usage'];
        $this->batchSize = $options['batch_size'];
        $this->memoryMonitoringEnabled = $options['memory_monitoring_enabled'];
    }
    
    /**
     * 处理流式数据
     * 
     * @param Generator $stream 流式数据
     * @param callable $processor 处理函数
     * @return Generator 处理结果
     */
    public function processStream(Generator $stream, callable $processor): Generator
    {
        $batch = [];
        $count = 0;
        
        foreach ($stream as $item) {
            $batch[] = $item;
            $count++;
            
            // 检查内存使用
            if ($this->memoryMonitoringEnabled && $this->isMemoryLimitExceeded()) {
                // 处理当前批次并清空
                if (!empty($batch)) {
                    foreach ($processor($batch) as $result) {
                        yield $result;
                    }
                    $batch = [];
                    $this->currentMemoryUsage = 0;
                }
            }
            
            // 检查批次大小
            if ($count >= $this->batchSize) {
                foreach ($processor($batch) as $result) {
                    yield $result;
                }
                $batch = [];
                $count = 0;
            }
        }
        
        // 处理剩余数据
        if (!empty($batch)) {
            foreach ($processor($batch) as $result) {
                yield $result;
            }
        }
    }
    
    /**
     * 处理流式消息
     * 
     * @param Generator $messageStream 消息流
     * @param callable $processor 处理函数
     * @return Generator 处理结果
     */
    public function processMessageStream(Generator $messageStream, callable $processor): Generator
    {
        $batch = [];
        $count = 0;
        
        foreach ($messageStream as $message) {
            // 估算消息大小
            $messageSize = $this->estimateMessageSize($message);
            $this->currentMemoryUsage += $messageSize;
            
            $batch[] = $message;
            $count++;
            
            // 检查内存使用
            if ($this->memoryMonitoringEnabled && $this->isMemoryLimitExceeded()) {
                // 处理当前批次并清空
                if (!empty($batch)) {
                    foreach ($processor($batch) as $result) {
                        yield $result;
                    }
                    
                    // 释放消息对象
                    $this->releaseMessages($batch);
                    $batch = [];
                    $this->currentMemoryUsage = 0;
                }
            }
            
            // 检查批次大小
            if ($count >= $this->batchSize) {
                foreach ($processor($batch) as $result) {
                    yield $result;
                }
                
                // 释放消息对象
                $this->releaseMessages($batch);
                $batch = [];
                $count = 0;
            }
        }
        
        // 处理剩余数据
        if (!empty($batch)) {
            foreach ($processor($batch) as $result) {
                yield $result;
            }
            
            // 释放消息对象
            $this->releaseMessages($batch);
        }
    }
    
    /**
     * 估算消息大小
     * 
     * @param Message $message 消息对象
     * @return int 估算大小（字节）
     */
    protected function estimateMessageSize(Message $message): int
    {
        try {
            return strlen($message->serializeToString());
        } catch (\Exception $e) {
            // 如果序列化失败，返回一个估算值
            return 1024;  // 1KB
        }
    }
    
    /**
     * 释放消息对象
     * 
     * @param array $messages 消息对象数组
     * @return void
     */
    protected function releaseMessages(array $messages): void
    {
        // 如果启用了消息池，归还对象到池
        if (class_exists('WebmanGrpc\\Serialization\\MessagePool')) {
            foreach ($messages as $message) {
                \WebmanGrpc\Serialization\MessagePool::returnMessage($message);
            }
        }
    }
    
    /**
     * 检查内存是否超过限制
     * 
     * @return bool 是否超过限制
     */
    protected function isMemoryLimitExceeded(): bool
    {
        return $this->currentMemoryUsage > $this->maxMemoryUsage;
    }
    
    /**
     * 创建流式读取器
     * 
     * @param callable $reader 读取函数
     * @param array $options 选项
     * @return Generator 流式数据
     */
    public static function createStreamReader(callable $reader, array $options = []): Generator
    {
        $defaultOptions = [
            'buffer_size' => 8192,
            'chunk_size' => 1024,
        ];
        
        $options = array_merge($defaultOptions, $options);
        
        $buffer = '';
        $resource = $reader();
        
        if (!is_resource($resource)) {
            return;
        }
        
        while (!feof($resource)) {
            $chunk = fread($resource, $options['chunk_size']);
            if ($chunk === false) {
                break;
            }
            
            $buffer .= $chunk;
            
            // 当缓冲区达到一定大小时，yield数据
            if (strlen($buffer) >= $options['buffer_size']) {
                yield $buffer;
                $buffer = '';
            }
        }
        
        // yield剩余数据
        if (!empty($buffer)) {
            yield $buffer;
        }
        
        fclose($resource);
    }
    
    /**
     * 创建流式写入器
     * 
     * @param callable $writer 写入函数
     * @param array $options 选项
     * @return callable 写入函数
     */
    public static function createStreamWriter(callable $writer, array $options = []): callable
    {
        $defaultOptions = [
            'buffer_size' => 8192,
            'auto_flush' => true,
        ];
        
        $options = array_merge($defaultOptions, $options);
        
        $resource = $writer();
        $buffer = '';
        
        return function ($data) use ($resource, &$buffer, $options) {
            $buffer .= $data;
            
            // 当缓冲区达到一定大小时，写入数据
            if (strlen($buffer) >= $options['buffer_size'] || $options['auto_flush']) {
                fwrite($resource, $buffer);
                $buffer = '';
            }
            
            return true;
        };
    }
    
    /**
     * 创建内存高效的迭代器
     * 
     * @param array $data 数据数组
     * @param int $chunkSize 块大小
     * @return Generator 分块数据
     */
    public static function createMemoryEfficientIterator(array $data, int $chunkSize = 100): Generator
    {
        $chunk = [];
        
        foreach ($data as $item) {
            $chunk[] = $item;
            
            if (count($chunk) >= $chunkSize) {
                yield $chunk;
                $chunk = [];
            }
        }
        
        if (!empty($chunk)) {
            yield $chunk;
        }
    }
    
    /**
     * 获取当前内存使用量
     * 
     * @return int 内存使用量（字节）
     */
    public function getCurrentMemoryUsage(): int
    {
        return $this->currentMemoryUsage;
    }
    
    /**
     * 获取最大内存使用量
     * 
     * @return int 最大内存使用量（字节）
     */
    public function getMaxMemoryUsage(): int
    {
        return $this->maxMemoryUsage;
    }
    
    /**
     * 设置最大内存使用量
     * 
     * @param int $maxMemoryUsage 最大内存使用量（字节）
     * @return void
     */
    public function setMaxMemoryUsage(int $maxMemoryUsage): void
    {
        $this->maxMemoryUsage = $maxMemoryUsage;
    }
    
    /**
     * 启用内存监控
     * 
     * @return void
     */
    public function enableMemoryMonitoring(): void
    {
        $this->memoryMonitoringEnabled = true;
    }
    
    /**
     * 禁用内存监控
     * 
     * @return void
     */
    public function disableMemoryMonitoring(): void
    {
        $this->memoryMonitoringEnabled = false;
    }
}